"""
Case Type   : 极致RTO支持备机可读
Case Name   : 资源池化网络复制双集群，备集群开启极致RTO，主备集群间切换场景下验证备集群可读
Create At   : 2025/04
Owner       : haomeng
Description :
    1、主备集群内均开启极致RTO特性，重启使其生效
    2、主集群内创建行存段页式range分区表，插入数据
    3、备集群首备节点查看表数据
    4、备集群从备节点查看表数据
    5、主备集群内同时执行gs_ddr switchover切换使主备集群间进行切换，查看集群状态
    6、在新的备集群首备节点内查看原始表数据
    7、在新的备集群从备节点查看原始表数据
    8、在新主集群点内创建表，插入数据
    9、在新备集群内首备查看新创建的段页式表数据
    10、在新备集群内从备查看新创建的段页式表数据
    11、清理环境，恢复环境初始状态
Expect      :
    1、设置参数成功，重启成功
    2、创建成功，插入数据成功
    3、查看成功，与主集群内查看数据一致
    4、查看失败，从备节点不可读
    5、主备集群间切换成功，集群状态正常
    6、查看成功
    7、查看失败不可读
    8、创建成功，插入数据成功
    9、查看成功
    10、查看失败不可读
    11、清理成功，恢复环境成功
History     :
"""

import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonThread import CommonThread
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class DisasterRTOStandbyRead(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.p_comsh = [CommonSH(n) for n in ('P_PrimaryDbUser', 'P_Standby1DbUser', 'P_Standby2DbUser')]
        self.s_comsh = [CommonSH(n) for n in ('S_PrimaryDbUser', 'S_Standby1DbUser', 'S_Standby2DbUser')]
        self.user_node = [Node('P_PrimaryDbUser'), Node('S_PrimaryDbUser')]
        self.constant = Constant()

        self.param_list = ['recovery_parse_workers', 'recovery_redo_workers', 'ss_disaster_mode']
        self.value_list = ['4', '4']
        self.default_value = []
        self.table_list = ['ss_rto_009', 'ss_rto_009_new', 'new_ss_rto_009']
        self.pri_switch = f"gs_ddr -t switchover -m disaster_standby"
        self.sta_switch = f"gs_ddr -t switchover -m primary"

        self.log.info('---校验当前环境是否为网络复制双集群---')
        mode = self.p_comsh[0].execut_db_sql(f'show {self.param_list[2]};')
        self.log.info(mode)
        self.assertIn('stream', mode, '当前环境不是网络复制双集群')

    def test_rto_standby_read(self):
        text = '-----step1:主备集群内均开启极致RTO特性，重启使其生效;  expect:设置成功，重启成功-----'
        self.log.info(text)

        for param in self.param_list[:2]:
            value = self.s_comsh[0].execut_db_sql(f'show {param}')
            self.default_value.append(value.splitlines()[-2].strip())
        self.log.info(self.default_value)

        for param in self.param_list[:2]:
            pri_set_param = self.p_comsh[0].execute_gsguc('set', self.constant.GSGUC_SUCCESS_MSG,
                                                    f'{param} = {self.value_list[self.param_list.index(param)]}')
            self.log.info(pri_set_param)
            self.assertTrue(pri_set_param, '主集群参数设置失败')

            sta_set_param = self.s_comsh[0].execute_gsguc('set', self.constant.GSGUC_SUCCESS_MSG,
                                                    f'{param} = {self.value_list[self.param_list.index(param)]}')
            self.log.info(sta_set_param)
            self.assertTrue(sta_set_param, '备集群参数设置失败')

        self.log.info('---重启集群使参数生效---')
        pri_restart = self.p_comsh[0].restart_db_cluster()
        self.log.info(pri_restart)
        self.assertTrue(pri_restart, '备集群集群重启失败')
        sta_restart = self.s_comsh[0].restart_db_cluster()
        self.log.info(sta_restart)
        self.assertTrue(sta_restart, '备集群集群重启失败')

        self.log.info('---查看参数值修改是否成功---')
        for param in self.param_list[:2]:
            pri_value = self.p_comsh[0].execut_db_sql(f'show {param};')
            self.assertEqual(pri_value.splitlines()[-2].strip(),
                             self.value_list[self.param_list.index(param)], '参数未修改成功')

            sta_value = self.s_comsh[0].execut_db_sql(f'show {param};')
            self.assertEqual(sta_value.splitlines()[-2].strip(),
                             self.value_list[self.param_list.index(param)], '参数未修改成功')

        text = '-----step2:主集群内创建行存段页式range分区表，插入数据  expect:创建成功，插入数据成功-----'
        self.log.info(text)

        sql1 = f"drop table if exists {self.table_list[0]};" \
               f"create table {self.table_list[0]}(id int, name varchar(200), age int, tel int) with(segment=on) " \
               f"partition by range(tel) " \
               f"(partition p1 values less than(5000), partition p2 values less than(10000), " \
               f"partition p3 values less than(15000), partition p4 values less than(20000), " \
               f"partition p5 values less than(25000), partition p6 values less than(30000), " \
               f"partition p7 values less than(40000), partition p8 values less than(50000)); " \
               f"declare    " \
               f"    i integer;" \
               f"begin" \
               f"    for i in 1..100000 loop" \
               f"    insert into {self.table_list[0]} values(random()*100,'name_'||random()*10,random()*100,random()*10000);" \
               f"    end loop;" \
               f"end;" \
               f"alter table {self.table_list[0]} rename to {self.table_list[1]};" \
               f"alter table {self.table_list[1]} add new_col1 int, add new_col2 text first, add new_col3 float after new_col2;" \
               f"update {self.table_list[1]} set tel = tel * 2 where id%2 != 0 and id < 1000;" \
               f"delete from {self.table_list[1]} where id < 20;"
        self.log.info(sql1)

        res1 = self.p_comsh[0].execut_db_sql(sql1)
        self.log.info(res1)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, res1, '建表失败')
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG, res1, '匿名块执行失败，插入数据失败')
        self.assertIn(self.constant.ALTER_TABLE_MSG, res1, 'alter表失败')
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, res1, 'update表数据失败')
        self.assertIn(self.constant.DELETE_SUCCESS_MSG, res1, 'delete表数据失败')

        text = '-----step3:备集群首备节点查看表数据  expect:查看成功-----'
        self.log.info(text)

        self.log.info('---等待备集群数据同步---')
        flag = self.s_comsh[0].check_data_consistency()
        self.assertTrue(flag, '数据同步失败')

        sql2 = f"select count(*) from {self.table_new};"
        self.log.info(sql2)
        select_res = self.s_comsh[0].execut_db_sql(sql2)
        self.log.info(select_res)

        p_select = self.p_comsh[0].execut_db_sql(sql2)
        self.log.info(p_select)
        self.assertEqual(p_select.splitlines()[-2].strip(), select_res.splitlines()[-2].strip(), '查询结果不一致')

        text = '-----step4:备集群从备节点查看表数据  expect:查看失败,从备节点不可读-----'
        self.log.info(text)

        ss_select = self.s_comsh[1].execut_db_sql(sql2)
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

        text = '-----step5:主备集群内同时执行switchover切换进行集群间切换，查看集群状态  ' \
               'expect:主集群与备集群切换成功，集群状态正常-----'
        self.log.info(text)

        self.log.info('---主集群执行switchover---')
        switch_exec1 = CommonThread(self.user_node[0], args=(self.pri_switch,))
        switch_exec1.setDaemon(True)
        switch_exec1.start()

        self.log.info('---备集群执行switchover---')
        switch_exec2 = CommonThread(self.user_node[1], args=(self.sta_switch,))
        switch_exec2.setDaemon(True)
        switch_exec2.start()

        switch_exec1.join(10 * 60)
        ret1 = switch_exec1.get_result()
        self.log.info(ret1)
        self.assertIn('Successfully do dorado disaster recovery switchover',
                      ret1, '主集群降备失败')

        switch_exec2.join(10 * 60)
        ret2 = switch_exec2.get_result()
        self.log.info(ret2)
        self.assertIn('Successfully do dorado disaster recovery switchover',
                      ret2, '备集群升主失败')

        self.log.info('---查看新主集群状态---')
        status1 = self.s_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(status1)
        self.assertIn('cluster_state  : Normal', status1, '集群状态异常')
        self.assertIn('Primary Normal', status1.splitlines()[-1], '执行失败' + text)

        self.log.info('---查看新备集群状态---')
        status2 = self.p_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(status2)
        self.assertIn('cluster_state  : Normal', status2, '集群状态异常')
        self.assertIn('Main Standby Normal', status2.splitlines()[-1], '执行失败' + text)

        text = '-----step6:在新的首备节点内查看原始表数据  expect:查看成功-----'
        self.log.info(text)

        self.log.info('---等待新的首备节点数据同步---')
        flag = self.p_comsh[0].check_data_consistency()
        self.assertTrue(flag, '数据同步失败')

        new_s_select = self.p_comsh[0].execut_db_sql(sql2)
        self.log.info(new_s_select)
        self.assertEqual(new_s_select.splitlines()[-2].strip(),
                         select_res.splitlines()[-2].strip(), '查询结果不一致')

        text = '-----step7:在新的从备节点查看表数据  expect:查看失败,从备节点不可读-----'
        self.log.info(text)

        ss_select = self.p_comsh[1].execut_db_sql(sql2)
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

        text = '-----step8:在新主集群点内创建表，插入数据  expect:创建成功，插入数据成功-----'
        self.log.info(text)
        sql3 = f"drop table if exists {self.table_list[2]};" \
               f"create table {self.table_list[2]}(id int, name varchar(200), age int, tel int) with(segment=on) " \
               f"partition by range(tel) " \
               f"(partition p1 values less than(5000), partition p2 values less than(10000), " \
               f"partition p3 values less than(15000), partition p4 values less than(20000), " \
               f"partition p5 values less than(25000), partition p6 values less than(30000), " \
               f"partition p7 values less than(40000), partition p8 values less than(50000)); " \
               f"declare    " \
               f"    i integer;" \
               f"begin" \
               f"    for i in 1..100000 loop" \
               f"    insert into {self.table_list[2]} values(random()*100,'name_'||random()*10,random()*100,random()*10000);" \
               f"    end loop;" \
               f"end;" \
               f"alter table {self.table_list[2]} rename to {self.table_list[0]};" \
               f"alter table {self.table_list[0]} add new_col1 int, add new_col2 text first, add new_col3 float after new_col2;" \
               f"update {self.table_list[0]} set tel = tel * 2 where id%2!= 0 and id < 1000;" \
               f"delete from {self.table_list[0]} where id < 20;"
        self.log.info(sql3)
        res3 = self.s_comsh[0].execut_db_sql(sql3)
        self.log.info(res3)
        self.assertIn(self.constant.CREATE_TABLE_SUCCESS, res3, '建表失败')
        self.assertIn(self.constant.CREATE_ANONYMOUS_BLOCK_SUCCESS_MSG, res3, '匿名块执行失败，插入数据失败')
        self.assertIn(self.constant.ALTER_TABLE_MSG, res3, 'alter表失败')
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, res3, 'update表数据失败')
        self.assertIn(self.constant.DELETE_SUCCESS_MSG, res3, 'delete表数据失败')

        text = '-----step9:在新备集群内首备查看新创建的段页式表数据  expect:查看成功-----'
        self.log.info(text)

        self.log.info('---等待新的首备节点数据同步---')
        flag = self.p_comsh[0].check_data_consistency()
        self.assertTrue(flag, '数据同步失败')

        new_s_select = self.p_comsh[0].execut_db_sql(sql2)
        self.log.info(new_s_select)

        new_p_select = self.s_comsh[0].execut_db_sql(sql2)
        self.log.info(new_p_select)
        self.assertEqual(new_p_select.splitlines()[-2].strip(),
                         new_s_select.splitlines()[-2].strip(), '查询结果不一致')

        text = '-----step10:在新备集群内从备查看新创建的段页式表数据  expect:查看失败,从备节点不可读-----'
        self.log.info(text)

        ss_select = self.p_comsh[1].execut_db_sql(sql2)
        self.log.info(ss_select)
        self.assertIn(self.constant.disaster_read_fail_msg, ss_select, '从备节点查询成功')

    def tearDown(self):
        text = '-----step11:清理环境，恢复环境初始状态  expect:清理成功，恢复环境成功-----'
        self.log.info(text)

        self.log.info('-----清理表-----')
        for i in self.table_list[:2]:
            drop = f"drop table if exists {i} cascade;"
            self.log.info(drop)
            self.drop_res = self.s_comsh[0].execut_db_sql(drop)
            self.log.info(self.drop_res)

        self.log.info('-----恢复环境默认参数-----')
        for param in self.param_list[:2]:
            p_reset_param = self.s_comsh[0].execute_gsguc('set',
                                                      self.constant.GSGUC_SUCCESS_MSG,
                                                      f'{param} = {self.default_value[self.param_list.index(param)]}')
            self.log.info(p_reset_param)

            s_reset_param = self.p_comsh[0].execute_gsguc('set',
                                                      self.constant.GSGUC_SUCCESS_MSG,
                                                      f'{param} = {self.default_value[self.param_list.index(param)]}')
            self.log.info(s_reset_param)

        self.log.info('---新主集群执行switchover---')
        switch_exec3 = CommonThread(self.user_node[1], args=(self.sta_switch,))
        switch_exec3.setDaemon(True)
        switch_exec3.start()

        self.log.info('---新备集群执行switchover---')
        switch_exec4 = CommonThread(self.user_node[0], args=(self.pri_switch,))
        switch_exec4.setDaemon(True)
        switch_exec4.start()

        switch_exec3.join(10 * 60)
        ret3 = switch_exec3.get_result()
        self.log.info(ret3)

        switch_exec4.join(10 * 60)
        ret4 = switch_exec4.get_result()
        self.log.info(ret4)

        self.log.info('-----查看参数默认值是否恢复-----')
        for param in self.param_list[:2]:
            reset_value = self.s_comsh[0].execut_db_sql(f'show {param};')
            self.log.info(reset_value)
            self.assertEqual(reset_value.splitlines()[-2].strip(),
                             self.default_value[self.param_list.index(param)], '参数恢复失败')

        status = self.s_comsh[0].exec_cm_ctl('query', '-Cv')
        self.log.info(status)

        self.assertIn('cluster_state  : Normal', status, '集群状态异常')
        self.assertIn('Main Standby Normal', status, '执行失败' + text)
        self.assertEqual(2, self.drop_res.count(self.constant.DROP_TABLE_SUCCESS), '删除表失败')
        self.assertIn('Successfully do dorado disaster recovery switchover',
                      ret3, '主集群降备失败')
        self.assertIn('Successfully do dorado disaster recovery switchover',
                      ret4, '备集群升主失败')
        self.log.info(f'----{os.path.basename(__file__)} end----')
